package per.chao.mqtt.domain.protocol.pubrec;

import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import per.chao.mqtt.domain.protocol.BaseProtocolHandler;
import per.chao.mqtt.domain.repo.PubRelMessageRepo;
import per.chao.mqtt.domain.repo.PublishMessageRepo;
import per.chao.mqtt.domain.shared.store.DupPubRelMessageStore;

/**
 * Description:
 *
 * @author W.Chao
 * @date 2021/6/19 23:05
 **/
@Component
@Slf4j
public class PubRec extends BaseProtocolHandler {

	@Autowired
	private PublishMessageRepo publishMessageRepo;
	@Autowired
	private PubRelMessageRepo pubRelMessageRepo;

	@Override
	public void handleMessage(Channel channel, Object msg) {
		MqttMessageIdVariableHeader mqttMsgIdVariableHeader = (MqttMessageIdVariableHeader) msg;
		int messageId = mqttMsgIdVariableHeader.messageId();
		String clientId = (String) channel.attr(AttributeKey.valueOf("clientId")).get();
		log.info("PUBREC - clientId: {}, messageId: {}", clientId, messageId);
		MqttMessage pubRecMessage = MqttMessageFactory.newMessage(
				new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_MOST_ONCE, false, 0),
				MqttMessageIdVariableHeader.from(messageId),
				null
		);
		publishMessageRepo.remove(clientId, messageId);
		DupPubRelMessageStore dupPubRelMessageStore = new DupPubRelMessageStore();
		dupPubRelMessageStore.setClientId(clientId);
		dupPubRelMessageStore.setMessageId(messageId);
		pubRelMessageRepo.put(clientId, dupPubRelMessageStore);
		channel.writeAndFlush(pubRecMessage);
	}
}
